package com.wyh.apitest.source.CustomSource

import com.wyh.apitest.source.SensorReading
import org.apache.flink.streaming.api.functions.source.SourceFunction

import scala.util.Random

//自定义Source
class MySensorSource extends SourceFunction[SensorReading] {
  //定义一个标志位，用来表示数据源是否正常运行发出数据
  var running: Boolean = true

  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {

    //定义一个随机数发生器
    val random = new Random()

    //随机生成一组传感器的初始温度,因为之后温度会改变，所以定义成var,高斯随机数
    var curTemp = 1.to(10).map(i => {
      ("sensor_" + i, random.nextGaussian() * 100)
    })

    //定义一个无限循环，不停地产生数据，除非被cancel
    while (running) {

      //在上次基础上更新温度值
      curTemp = curTemp.map(t => {
        (t._1, t._2 + random.nextGaussian() * 100)
      })

      //获取当前时间戳，加入到数据中
      var curTime = System.currentTimeMillis()

      curTemp.foreach(t => {
        sourceContext.collect(SensorReading(t._1, curTime, t._2))
      })


      println("------------------------------------------------ ")
      Thread.sleep(1000)


    }

  }

  override def cancel(): Unit = {
    running = false


  }
}
